package com.briup.prs;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.lib.db.DBConfiguration;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.db.DBOutputFormat;
import org.apache.hadoop.mapreduce.lib.db.DBWritable;
import org.apache.hadoop.mapreduce.lib.input.KeyValueLineRecordReader;
import org.apache.hadoop.mapreduce.lib.input.KeyValueTextInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;

/**
 * @author adam
 * @date 2022/2/10
 */
public class Step9 extends Configured implements Tool {
    public static class RecommendShop implements DBWritable, Writable {
        private Long user_id;
        private String shop_id;
        private Double recommand_value;

        public RecommendShop(Long user_id, String shop_id, Double recommand_value) {
            this.user_id = user_id;
            this.shop_id = shop_id;
            this.recommand_value = recommand_value;
        }

        public RecommendShop() {
        }

        @Override
        public String toString() {
            return "RecommendShop{" +
                    "user_id='" + user_id + '\'' +
                    ", shop_id='" + shop_id + '\'' +
                    ", recommand_value=" + recommand_value +
                    '}';
        }

        @Override
        public void write(PreparedStatement statement) throws SQLException {

            statement.setLong(1, user_id);
            statement.setString(2, shop_id);
            statement.setDouble(3, recommand_value);
        }

        @Override
        public void readFields(ResultSet resultSet) throws SQLException {

            user_id = resultSet.getLong("user_id");
            shop_id = resultSet.getString("shop_id");
            recommand_value = resultSet.getDouble("recommand_value");

        }

        @Override
        public void write(DataOutput out) throws IOException {
            out.writeLong(user_id);
            out.writeUTF(shop_id);
            out.writeDouble(recommand_value);
        }

        @Override
        public void readFields(DataInput in) throws IOException {

            this.user_id = in.readLong();
            this.shop_id = in.readUTF();
            this.recommand_value = in.readDouble();
        }
    }

    public static class S9Mapper extends Mapper<Text, Text, Text, Text> {
        @Override
//        key=用户:商品   value:推荐值
        protected void map(Text key, Text value, Context context) throws IOException, InterruptedException {
            context.write(key, value);
        }
    }

    public static class S9Reducer extends Reducer<Text, Text, RecommendShop, NullWritable> {

        @Override
//        key=用户:商品   value:推荐值
        protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
//            data=[用户,商品]
            String[] data = key.toString().split(":");
            Long userId = Long.parseLong(data[0]);
            String shopId = data[1];
            Double recommandValue =null;
            for (Text value : values) {
                recommandValue=Double.parseDouble(value.toString());
                System.out.println(key.toString()+"-------------"+recommandValue);
            }

            RecommendShop recommendShop = new RecommendShop(userId, shopId, recommandValue);
            context.write(recommendShop,NullWritable.get());

        }
    }

    @Override
    public int run(String[] args) throws Exception {
        Configuration conf = new Configuration();
        String in = conf.get("in");
        conf.set(KeyValueLineRecordReader.KEY_VALUE_SEPARATOR,"\t");
        DBConfiguration.configureDB(conf, "com.mysql.cj.jdbc.Driver",
                "jdbc:mysql://127.0.0.1:3306/shops", "root", "root");
        Job job = Job.getInstance(conf, "step9");
        job.setJarByClass(Step9.class);
        // Mapper
        job.setMapperClass(S9Mapper.class);
        KeyValueTextInputFormat.addInputPath(job, new Path(in));
        job.setInputFormatClass(KeyValueTextInputFormat.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);


        job.setReducerClass(S9Reducer.class);
        job.setOutputFormatClass(DBOutputFormat.class);
        // 输出到哪些表、字段
        DBOutputFormat.setOutput(job, "t_recommend_shop",  "user_id", "shops","recommend_value");
        // 输出格式
        job.setOutputKeyClass(RecommendShop.class);
        job.setOutputValueClass(NullWritable.class);

        job.waitForCompletion(true);

        return 0;
    }

    public static void main(String[] args) throws Exception {
        ToolRunner.run(new Step9(),args);
    }
}
